namespace Stee.CAP8.MessageBus.Sagas.Payment.Services
{
    using Stee.CAP8.MessageBus.Interface.Interfaces;
    using Stee.CAP8.MessageBus.Interface.Models;
    using Stee.CAP8.MessageBus.Kafka.Publisher;
    using Stee.CAP8.MessageBus.Kafka.Subscriber;
    using Stee.CAP8.MessageBus.Sagas.Core.Common;
    using Stee.CAP8.MessageBus.Sagas.Payment.Dtos;
    using System;
    using System.Threading;
    using System.Threading.Tasks;



    public class PaymentService
    {
        private IPublisher _publisher;
        private ISubscriber _subscriber;
        public PaymentService()
        {
        }

        public void Subscirbie()
        {
            this._publisher = new PublisherImpl();
            this._subscriber = new SubscriberImpl();
            this._subscriber.Initialize(new TopicName("payment_application", "payment_application"),
                this.OnCommandReceived,
                (errorCode, isFatal) => Console.WriteLine(errorCode));
            if (this._subscriber.IsInitialized)
            {
                this._subscriber.Subscribe();
            }

        }

        private bool OnCommandReceived(TopicName topicName, CommandMessage message)
        {

            try
            {
                #region Check & Log
                if (topicName == null)
                {
                    throw new ArgumentNullException(nameof(topicName));
                }
                else if (message == null)
                {
                    throw new ArgumentNullException(nameof(message));
                }

                //// Consume command message type
                string output = $"Consumed command message <CommandMessage>: {message}";

                Console.WriteLine(output);
                Console.WriteLine("Message consumed successfully.");
                #endregion


                if (message.Action == "PaymentSub.Payment")
                {
                    var data = message.GetObject<PaymentInfo>();
                    if (data == null)
                    {
                        throw new Exception("传入错误的数据类型");
                    }
                    var paymentResult = this.Payment(data);
                    if (!paymentResult.IsCompleted)
                    {
                        paymentResult.Wait();
                    }
                    message.SetObjectType(data);
                }
                this.ReplyCommand(message);

                return true;
            }
            catch (Exception ex)
            {
                Console.WriteLine(ex.Message);
                throw;
            }
        }

        //改为paymentInfo
        public Task Payment(PaymentInfo payinfo)
        {

            Console.WriteLine("收到支付信息" + payinfo.PaymentMethod + "||" + payinfo.Amount);
            Console.WriteLine("支付操作");
            payinfo.PaymentMethod += payinfo.PaymentMethod;
            return Task.CompletedTask;

        }

  

        /// <summary>
        /// Reply Saga Command
        /// </summary>
        /// <param name="message"></param>
        private void ReplyCommand(CommandMessage message)
        {
            this._publisher.Initialize(publishOutcome => publishOutcome != null, (errorCode, isFatal) => Console.WriteLine(errorCode));
            //replay信息
            var topicInfo = new TopicName(message.ReplyChannel, message.ReplyChannel);
            if (this._publisher.IsInitialized)
            {
                var objectPublishToken = new CancellationTokenSource();
                if (this._publisher.PublishCommand(topicInfo, message, objectPublishToken.Token))
                {
                    var msg = $"Publish CommandMessage object => {message}";
                    Console.WriteLine(msg);
                }
                else
                {
                    var msg = $"[Error] Publish CommandMessage object failed => {message}";
                    Console.WriteLine(msg);
                }
            }
        }

    }






}
